package com.aotain.nyx.cc;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.nyx.common.URLStatTuple;
import com.aotain.nyx.util.FlowDirection;


/**
 * 统计一个源IP访问的相同URL的个数的Map
 * @author Administrator
 *
 */
public class URLStatMap  implements MapFunction<Tuple2<String,String>,Tuple2<String,URLStatTuple>>{

	/**
	 * 
	 */
	private static final long serialVersionUID = 7735484456076574287L;

	@Override
	public Tuple2<String, URLStatTuple> map(Tuple2<String, String> arg0)
			throws Exception {
		// TODO Auto-generated method stub
//		// [0-4]dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" + 
		// [5-9]domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" + 
		// [10-14]upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" + 
		// [15-19]dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post | OUT/IN
		
		Tuple2<String, URLStatTuple> ret = new Tuple2<String, URLStatTuple>();
		String logString = arg0.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		String dip = items[1];
		String sip = items[2];
		String gis = items[4];
		String idc = items[9];
		String useragent = items[13];
		String cookie = items[14];
		String flowdirection = items[19];
		
		String url = items[6];
		
		if(url.isEmpty())
			return ret;
		
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET
//		long dnStreamOctets = 0; //DNSTREAMOCTETS
//		long dnStreamPacket = 0; //DNSTREAMPACKET
		
		String tripleKey = dip + "|" + sip +"|" + url + "|" 
				+ gis + "|" + idc + "|" + useragent + "|" + cookie;
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
//		dnStreamOctets = Long.parseLong(items[15]);
//		dnStreamPacket = Long.parseLong(items[16]);
		
		URLStatTuple info = new URLStatTuple();
		info.setUpStreamOctets(upStreamOctets);
		info.setUpStreamPacket(upStreamPacket);
		info.setHTTPNum(1l);
		info.setUrl(url);
		info.setGIS(gis);
	    info.setDestIP(dip);
	    info.setSourceIP(sip);
	    info.setFlowDirection(flowdirection);
		String retKey = String.format("%s", tripleKey);
		
		
		
		ret.setFields(retKey, info);
		return ret;
	}

}
